package com.codeduck.demo.withjava;

import com.codeduck.constant.KafkaConst;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.*;

/**
 * @author: <a href="mailto:codeduck@163.com">codeduck</a>
 * @date: 2023/01/14
 */
@RestController
@Slf4j
public class MsgController {

    @Autowired
    private KafkaTemplate<Object, Object> template;


    private static final Integer  cpuNum = Runtime.getRuntime().availableProcessors();
    @GetMapping(path = "/send/{what}")
    public void sendFoo(@PathVariable String what) throws ExecutionException, InterruptedException {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(cpuNum, cpuNum * 2, 0,TimeUnit.MILLISECONDS, new LinkedBlockingDeque<>(100));
        for (int i = 0; i < 10000; i++) {
            Callable<String> callable  = sendMsgToJavaTopic(what);
            Callable<String> callable1  = sendMsgToSimpleTopic(what);

            Future<String> res = executor.submit(callable);
            executor.submit(callable1);
            log.info("消息已发送 【{}】", res.get());
        }
        executor.shutdown();
    }

    private Callable<String> sendMsgToJavaTopic(String what) {
        return () -> {
            TimeUnit.MILLISECONDS.sleep(500);
            template.send(KafkaConst.JAVA_TOPIC, what);
            return what;
        };
    }

    private Callable<String> sendMsgToSimpleTopic(String what) {
        return () -> {
            template.send(KafkaConst.SIMPLE_TOPIC, what);
            return what;
        };
    }
}
